package indi.yangshenhui.hbase.dao.pool;

import indi.yangshenhui.hbase.dao.constant.HBaseConstant;
import indi.yangshenhui.hbase.dao.exception.HBaseDaoException;
import indi.yangshenhui.tool.ObjectUtil;
import indi.yangshenhui.tool.StringUtil;

import java.io.IOException;
import java.util.HashMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;

/**
 * 管理HBase的Configuration HTablePool
 * 
 * @author yangshenhui
 */
public class HTablePoolEngine {
	private static volatile HashMap<String, HTablePool> hTablePoolCache = new HashMap<String, HTablePool>();
	private static volatile HashMap<String, Configuration> configurationCache = new HashMap<String, Configuration>();

	public static Configuration createHBaseConfiguration(
			String hBaseInstanceName, Configuration configuration)
			throws HBaseDaoException {
		if (StringUtil.isEmpty(hBaseInstanceName)
				|| ObjectUtil.isNull(configuration)) {
			return null;
		}
		synchronized (configurationCache) {
			if (checkConfiguration(configuration)) {
				configurationCache.put(hBaseInstanceName, configuration);
			}
		}
		return configuration;
	}

	public static Configuration createHBaseConfiguration(
			String hBaseInstanceName, String hBaseZookeeperQuorum)
			throws HBaseDaoException {
		if (StringUtil.isEmpty(hBaseInstanceName)
				|| StringUtil.isEmpty(hBaseZookeeperQuorum)) {
			return null;
		}
		return createHBaseConfiguration(hBaseInstanceName,
				hBaseZookeeperQuorum,
				HBaseConstant.HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT);
	}

	public static Configuration createHBaseConfiguration(
			String hBaseInstanceName, String hBaseZookeeperQuorum,
			String hBaseZookeeperPort) throws HBaseDaoException {
		if (StringUtil.isEmpty(hBaseInstanceName)
				|| StringUtil.isEmpty(hBaseZookeeperQuorum)
				|| StringUtil.isEmpty(hBaseZookeeperPort)) {
			return null;
		}
		Configuration configuration = HBaseConfiguration.create();
		configuration.set("hbase.zookeeper.quorum", hBaseZookeeperQuorum);
		configuration.set("hbase.zookeeper.property.clientPort",
				hBaseZookeeperPort);
		configuration.set("ipc.ping.interval", HBaseConstant.IPC_PING_INTERVAL);
		configuration.setBoolean("hbase.ipc.client.tcpnodelay",
				HBaseConstant.HBASE_IPC_CLIENT_TCPNODELAY);
		configuration.set("hbase.client.pause",
				HBaseConstant.HBASE_CLIENT_PAUSE);
		configuration.set("hbase.client.retries.number",
				HBaseConstant.HBASE_CLIENT_RETRIES_NUMBER);
		synchronized (configurationCache) {
			if (checkConfiguration(configuration)) {
				configurationCache.put(hBaseInstanceName, configuration);
			}
		}
		return configuration;
	}

	public static Configuration getHBaseConfiguration()
			throws HBaseDaoException {
		return getHBaseConfiguration(HBaseConstant.HBASE_DEFAULT_INSTANCE);
	}

	public static Configuration getHBaseConfiguration(String hBaseInstanceName)
			throws HBaseDaoException {
		if (StringUtil.isEmpty(hBaseInstanceName)) {
			return null;
		}
		Configuration configuration = configurationCache.get(hBaseInstanceName);
		if ((ObjectUtil.isNull(configuration))
				&& (HBaseConstant.HBASE_DEFAULT_INSTANCE
						.equals(hBaseInstanceName))) {
			return createHBaseConfiguration(
					HBaseConstant.HBASE_DEFAULT_INSTANCE,
					HBaseConstant.HBASE_ZOOKEEPER_QUORUM,
					HBaseConstant.HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT);
		}
		return configuration;
	}

	public static HTablePool getHTablePool() throws HBaseDaoException {
		Configuration configuration = getHBaseConfiguration(HBaseConstant.HBASE_DEFAULT_INSTANCE);
		if (ObjectUtil.isNull(configuration)) {
			configuration = createHBaseConfiguration(
					HBaseConstant.HBASE_DEFAULT_INSTANCE,
					HBaseConstant.HBASE_ZOOKEEPER_QUORUM,
					HBaseConstant.HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT);
		}
		return getHTablePool(HBaseConstant.HBASE_DEFAULT_INSTANCE,
				HBaseConstant.HTABLE_POOL_MAX_SIZE);
	}

	public static HTablePool getHTablePool(String hBaseInstanceName, int maxSize)
			throws HBaseDaoException {
		if (StringUtil.isEmpty(hBaseInstanceName) || maxSize <= 0) {
			return null;
		}
		HTablePool hTablePool = hTablePoolCache.get(hBaseInstanceName);
		if (ObjectUtil.isNotNull(hTablePool)) {
			return hTablePool;
		}
		synchronized (hTablePoolCache) {
			hTablePool = hTablePoolCache.get(hBaseInstanceName);
			if (ObjectUtil.isNotNull(hTablePool)) {
				return hTablePool;
			}
			hTablePool = new HTablePool(
					getHBaseConfiguration(hBaseInstanceName), maxSize);
			hTablePoolCache.put(hBaseInstanceName, hTablePool);
		}
		return hTablePool;
	}

	public static HTableInterface getHTable(String tableName)
			throws HBaseDaoException {
		if (StringUtil.isEmpty(tableName)) {
			return null;
		}
		return getHTablePool().getTable(tableName);
	}

	public static HTableInterface getHTable(String tableName,
			String hBaseInstanceName) throws HBaseDaoException {
		if (StringUtil.isEmpty(tableName)
				|| StringUtil.isEmpty(hBaseInstanceName)) {
			return null;
		}
		return getHTablePool(hBaseInstanceName,
				HBaseConstant.HTABLE_POOL_MAX_SIZE).getTable(tableName);
	}

	public static HTableInterface createHTable(String tableName)
			throws HBaseDaoException {
		if (StringUtil.isEmpty(tableName)) {
			return null;
		}
		return createHTable(tableName, HBaseConstant.HBASE_DEFAULT_INSTANCE);
	}

	public static HTableInterface createHTable(String tableName,
			String hBaseInstanceName) throws HBaseDaoException {
		if (StringUtil.isEmpty(tableName)
				|| StringUtil.isEmpty(hBaseInstanceName)) {
			return null;
		}
		try {
			return new HTable(getHBaseConfiguration(hBaseInstanceName),
					tableName);
		} catch (IOException e) {
			throw new HBaseDaoException(e);
		}
	}

	public static void closeHTable(HTableInterface hTableInterface)
			throws HBaseDaoException {
		if (ObjectUtil.isNull(hTableInterface)) {
			return;
		}
		try {
			hTableInterface.flushCommits();
			hTableInterface.close();
		} catch (IOException e) {
			throw new HBaseDaoException(e);
		}

	}

	public static void optimizeHTable(HTableInterface hTableInterface,
			boolean autoFlush, boolean clearBufferOnFail, long writeBufferSize)
			throws HBaseDaoException {
		if (ObjectUtil.isNull(hTableInterface) || writeBufferSize <= 0) {
			return;
		}
		hTableInterface.setAutoFlush(autoFlush, clearBufferOnFail);
		try {
			hTableInterface.setWriteBufferSize(writeBufferSize);
		} catch (IOException e) {
			throw new HBaseDaoException(e);
		}
	}

	public static void optimizeHTable(HTableInterface hTableInterface,
			long writeBufferSize) throws HBaseDaoException {
		if (ObjectUtil.isNull(hTableInterface) || writeBufferSize <= 0) {
			return;
		}
		optimizeHTable(hTableInterface, false, false, writeBufferSize);
	}

	private static boolean checkConfiguration(Configuration configuration)
			throws HBaseDaoException {
		if (ObjectUtil.isNull(configuration)) {
			return false;
		}
		try {
			HBaseAdmin.checkHBaseAvailable(configuration);
		} catch (Exception e) {
			throw new HBaseDaoException(e);
		}
		return true;
	}

	private HTablePoolEngine() {
	}

}